// Copyright (c) 2020 Presto Labs Pte. Ltd.
// Author: donggu

#pragma once

#include <memory>
#include <string>
#include <unordered_set>
#include <utility>
#include <vector>

#include "coin2/exchange/base/order/external.h"
#include "coin2/exchange/base/order/processor.h"
#include "coin2/exchange/base/order/update_manager.h"

namespace coin2::exchange::base::order {

class ArchiveOrderSystem : public BaseOrderSubsystem, public executor::ArchiveMixin {
 public:
  ArchiveOrderSystem(
      const std::string& system_name,
      const MarketExchangeApi& mea,
      const di::ConstantInjection& ci,
      const OrderSystemConfig& config,
      base::executor::ArchiveSyncExecutor* archive_executor,
      IOrderSubscriber* callback,
      size_t max_record_size = 16 * 1024 * 1024)
      : BaseOrderSubsystem(system_name, config, callback),
        ArchiveMixin(archive_executor),
        buf_(new char[max_record_size]),
        record_(*reinterpret_cast<orio::record::TopicRecord*>(buf_.get())),
        upd_(system_name, mea),
        account_info_(mea) {
    const auto& archive = config.archive();
    interval_ = ::coin2::base::config::FromIntervalConfig(archive.interval());

    // product filter
    {
      auto interval = coin2::base::config::FromIntervalConfig(archive.interval());
      auto products = ProductRequestsFromConfig(mea.String(), config.products(), interval.first);
      for (auto& prod : products) {
        product_norms_.insert(prod->absolute_norm());
        account_info_.AddProduct(prod->absolute_norm());
        account_info_.AddCurrency(prod->base().currency());
        products_.push_back(std::move(prod));
        upd_.AddProduct(products_.back().get());
      }
    }

    AccountRequestProto request;
    request.set_owner(archive.owner());
    request.set_market_type(MarketType_Name(mea.market));
    request.set_exchange(ExchangeType_Name(mea.exchange));

    reader_ = std::make_unique<order_util::RawOrderLogReader>(
        request,
        archive.root_dir(),
        archive.machine(),
        interval_.first,   // begin
        interval_.second,  // end
        archive.max_record_size());
  }

  void Prepare() override {}
  void Cleanup() override {}

  /* ----------------------------- IOrderSubsytem ----------------------------- */
  bool is_supporting_separate_pos() const override {
    throw std::runtime_error("unreachable");
    return false;
  }
  BaseOrderGateway* gateway() override {
    throw std::runtime_error("unreachable");
  }  // common use case = submit/cancel order
  BaseOrderProcessor& processor() override {
    throw std::runtime_error("unreachable");
  }  // common use case = read order info

  /* ------------------------------ ArchiveMixin ----------------------------- */
  int64_t PeekTimestamp() override {
    if (!has_packet_) {
      while (reader_->Read(&record_)) {
        if (record_.timestamp < interval_.first) {
          continue;
        }

        // now: Topic -> parse to OrderLog proto -> publish to IStrategy
        // TODO(taekwon, xguo): Topic -> distributor -> gateway -> publish to IStrategy
        if (!proto_.ParseFromArray(record_.data, record_.data_len)) {
          throw std::runtime_error("[ArchiveOrderSystem] fails to parse proto");
        }

        if (proto_.type() == coin::proto::OrderGatewayLog::ORDER_EVENT) {
          if (product_norms_.find(proto_.event().symbol()) == product_norms_.end()) {
            continue;
          }
        }

        has_packet_ = true;
        break;
      }
      if (!has_packet_) {
        return 0;
      }
    }

    return record_.timestamp;
  }
  void ConsumeRecord() override {
    CHECK_THROW(has_packet_) << "Cannot consume a packet";
    has_packet_ = false;
    callback_->onOrderLog(proto_);

    UpdateOrderUpdate(record_.timestamp, proto_);
    if (upd_.is_account_info()) {
      callback_->onAccountInfo(upd_);
    } else if (upd_.is_product_order_info()) {
      callback_->onAccountOrder(upd_);
    } else {
      LOG(WARNING) << "Wrong OrderUpdate Type";
    }

    // archive_executor_->distributor()->OnTopicRecord(packet_.topic_name, &packet_);
  }

  void UpdateOrderUpdate(int64_t topic_timestamp, const coin::proto::OrderGatewayLog& log) {
    using LogType = OrderGatewayLog::LogType;
    // og_log file => OrderLog => OrderUpdate
    upd_.Reset();
    upd_.SetTimestamp(topic_timestamp);
    switch (log.type()) {
      case LogType::OrderGatewayLog_LogType_UNKNOWN_LOG_TYPE:
        throw std::runtime_error("unknown order type");
      case LogType::OrderGatewayLog_LogType_RESERVED_1:
        throw std::runtime_error("unknown order type");
      case LogType::OrderGatewayLog_LogType_ORDER_EVENT:
        product_order_info_.set_event(&log.event());
        upd_.SetProductOrderInfo(log.event().native_symbol(), &product_order_info_);
        break;
      case LogType::OrderGatewayLog_LogType_TICKER:
      case LogType::OrderGatewayLog_LogType_BALANCE:
        account_info_.UpdateAccountBalance(log.balance(), false);
        upd_.SetAccountInfo(&account_info_);
        break;
      case LogType::OrderGatewayLog_LogType_POSITION:
        account_info_.UpdateAccountPosition(log.position(), false);
        upd_.SetAccountInfo(&account_info_);
        break;
      case LogType::OrderGatewayLog_LogType_ORDER_BUNDLE:
      case LogType::OrderGatewayLog_LogType_FILL_BUNDLE:
        throw std::runtime_error("unknown order type");
      case LogType::OrderGatewayLog_LogType_ACCOUNT_REQUEST:
        log.account_request();
        throw std::runtime_error("unknown order type");
        break;
      case LogType::OrderGatewayLog_LogType_SUBMIT_REQUEST:
      case LogType::OrderGatewayLog_LogType_SUBMIT_RESPONSE:
      case LogType::OrderGatewayLog_LogType_AMEND_REQUEST:
      case LogType::OrderGatewayLog_LogType_AMEND_RESPONSE:
      case LogType::OrderGatewayLog_LogType_CANCEL_REQUEST:
      case LogType::OrderGatewayLog_LogType_CANCEL_RESPONSE:
      case LogType::OrderGatewayLog_LogType_PRIVATE_EXCHANGE_MESSAGE:
      default:
        throw std::runtime_error("unknown order type");
    }
  }

 protected:
  std::unique_ptr<order_util::RawOrderLogReader> reader_;
  bool has_packet_ = false;
  std::unique_ptr<char[]> buf_;
  orio::record::TopicRecord& record_;
  OrderGatewayLog proto_;
  std::pair<int64_t, int64_t> interval_;

  OrderUpdate upd_;
  AccountInfo account_info_;
  ProductOrderInfo product_order_info_;
  std::unordered_set<std::string> product_norms_;
  std::unordered_set<std::string> currency_norms_;
  std::vector<std::unique_ptr<IProduct>> products_;
};

}  // namespace coin2::exchange::base::order
